package cn.neo.hhmq.client;

import cn.kimmking.utils.HttpUtils;
import cn.kimmking.utils.ThreadUtils;
import cn.neo.hhmq.model.Message;
import cn.neo.hhmq.model.Result;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import lombok.Getter;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;

import java.util.ArrayList;
import java.util.List;

/**
 * broker for topics
 */

public class HHBroker {

    @Getter
    public static HHBroker Default = new HHBroker();

    public static String brokerUrl = "http://localhost:8765/hhmq";

    @Getter
    private MultiValueMap<String, HHConsumer<?>> consumers = new LinkedMultiValueMap<>();

    static {
        init();
    }

    private static void init() {
        ThreadUtils.getDefault().init(1);
        ThreadUtils.getDefault().schedule(() -> {
            MultiValueMap<String, HHConsumer<?>> consumers = Default.consumers;
            consumers.forEach((topic, consumers1) -> {
                consumers1.forEach(consumer -> {
                    Message<?> recv = consumer.recv(topic);
                    if(recv == null) return;
                   try {
                        consumer.getListener().onMessage(recv);
                        consumer.ack(topic, recv);
                    } catch (Exception ex) {
                       // TODO 消息失败重试
                   }
                });
            });
        }, 100, 100);
    }

    public HHProducer createProducer() {
        return new HHProducer(this);
    }

    public HHConsumer<?> createConsumer(String topic) {
        HHConsumer<?> hhConsumer = new HHConsumer<>(this);
        hhConsumer.sub(topic);
        return hhConsumer;
    }

    public void sub(String topic, String cid) {
        System.out.println(" ===> sub topic/cid:" + topic + "/" + cid);
        Result<String> result = HttpUtils.httpGet(
                brokerUrl + "/sub?t=" + topic + "&cid=" + cid,
                new TypeReference<Result<String>>() {
                });
        System.out.println(" ===> sub result: " + result);
    }

    public void unsub(String topic, String cid) {
        System.out.println(" ===> unsub topic/cid:" + topic + "/" + cid);
        Result<String> result = HttpUtils.httpGet(
                brokerUrl + "/unsub?t=" + topic + "&cid=" + cid,
                new TypeReference<Result<String>>() {
                });
        System.out.println(" ===> unsub result: " + result);
    }

    public boolean ack(String topic, String cid, int offset) {
        System.out.println(" ===> ack topic/cid/offset:" + topic + "/" + cid + "/" + offset);
        Result<String> result = HttpUtils.httpGet(
                brokerUrl + "/ack?t=" + topic + "&cid=" + cid + "&offset=" + offset,
                new TypeReference<Result<String>>() {
                });
        System.out.println(" ===> ack result: " + result);
        return result.getCode() == 1;
    }

    public Message<?> recv(String topic, String cid) {
        System.out.println(" ===> recv topic/cid:" + topic + "/" + cid);
        Result<Message<?>> result = HttpUtils.httpGet(
                brokerUrl + "/recv?t=" + topic + "&cid=" + cid,
                new TypeReference<Result<Message<?>>>() {
                });
        System.out.println(" ===> recv result: " + result);
        return result.getData();
    }

    public boolean send(String topic, Message message) {
        System.out.println(" ===> send topic/message:" + topic + "/" + message);
        System.out.println(JSON.toJSONString(message));
        Result<String> result = HttpUtils.httpPost(JSON.toJSONString(message),
                brokerUrl + "/send?t=" + topic,
                new TypeReference<Result<String>>() {
                });
        System.out.println(" ===> send result" + result);
        return result.getCode() == 1;
    }


    public void addConsumer(String topic, HHConsumer<?> consumer) {
        consumers.add(topic, consumer);
    }


    /**
     * 用于实现本地mq测试
     Map<String, HHMq> mqMapping = new ConcurrentHashMap<>();
     public HHMq createTopic(String topic) {
     return mqMapping.putIfAbsent(topic, new HHMq(topic));
     }

     public HHMq find(String topic) {
     return mqMapping.get(topic);
     }
     **/


}
